package conlon.rocketmq.proxy;

import conlon.rocketmq.mq.ProxyModel;
import conlon.rocketmq.utils.ProtoBufSerializer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.Objects;

/**
 * @Author conlon
 * @Description 消息发送的动态代理
 * @Date 2021/1/10 7:44
 **/
@Slf4j
@Component
public class ProducerDefaultProxy implements InvocationHandler {

    private final static String DEFAULT_PRODUCER_NAME = "defaultMqProducer";
    private final static String TRANSACTION_PRODUCER_NAME = "transactionMqProducer";

    private String topic;
    private String tags;
    private String producerName = DEFAULT_PRODUCER_NAME;

    private final DefaultMQProducer defaultMqProducer;

    private final TransactionMQProducer transactionMqProducer;

    private TransactionListener transactionListener;

    @Autowired
    public ProducerDefaultProxy(DefaultMQProducer defaultMqProducer, TransactionMQProducer transactionMqProducer) {
        this.defaultMqProducer = defaultMqProducer;
        this.transactionMqProducer = transactionMqProducer;
    }


    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        ProxyModel proxyModel = new ProxyModel()
                .setParameterTypes(method.getParameterTypes())
                .setClassName(method.getDeclaringClass().getName())
                .setMethodName(method.getName())
                .setArgs(args);
        Message message = new Message(topic, tags, ProtoBufSerializer.serialize(proxyModel));
        if (Objects.equals(producerName, DEFAULT_PRODUCER_NAME)) {
            SendResult result = defaultMqProducer.send(message);
            log.info("defaultMqProducer-invoke-SendResult : {} ", result);
        } else {
            if (!TransactionSynchronizationManager.isActualTransactionActive()) {
                throw new IllegalStateException(String.format("事务消息必须在事务中 : %s", message.toString()));
            }
            /**
             *  msg: 待投递的事务消息
             *  arg: 供本地事务执行程序使用的参数对象 对应 TransactionListen#executeLocalTransaction()
             **/
            SendResult result = transactionMqProducer.getDefaultMQProducerImpl().sendMessageInTransaction(message, transactionListener, null);
            log.info("transactionMqProducer-invoke-SendResult : {} ", result);
        }

        return null;
    }

    public <T> T getTransactionProxy(Class<T> tClass, String topic, String tags, TransactionListener transactionListener) {
        this.producerName = TRANSACTION_PRODUCER_NAME;
        this.transactionListener = transactionListener;
        return this.getProxy(tClass, topic, tags);
    }

    public <T> T getDefaultProxy(Class<T> tClass, String topic, String tags) {
        this.producerName = DEFAULT_PRODUCER_NAME;
        this.topic = topic;
        this.tags = tags;
        return tClass
                .cast(Proxy.newProxyInstance(tClass.getClassLoader(), new Class[]{tClass}, this));
    }

    private <T> T getProxy(Class<T> tClass, String topic, String tags) {
        this.topic = topic;
        this.tags = tags;
        return tClass
                .cast(Proxy.newProxyInstance(tClass.getClassLoader(), new Class[]{tClass}, this));
    }

    /**
     * TODO 事务同步器的具体解释
     public abstract class TransactionSynchronizationManager {

     //线程上下文中保存着【线程池对象：ConnectionHolder】的Map对象。线程可以通过该属性获取到同一个Connection对象。
     private static final ThreadLocal<Map<Object, Object>> resources = new NamedThreadLocal<>("Transactional resources");

     //事务同步器，是Spring交由程序员进行扩展的代码，每个线程可以注册N个事务同步器。
     private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations = new NamedThreadLocal<>("Transaction synchronizations");

     // 事务的名称
     private static final ThreadLocal<String> currentTransactionName = new NamedThreadLocal<>("Current transaction name");
     // 事务是否是只读
     private static final ThreadLocal<Boolean> currentTransactionReadOnly = new NamedThreadLocal<>("Current transaction read-only status");
     // 事务的隔离级别
     private static final ThreadLocal<Integer> currentTransactionIsolationLevel = new NamedThreadLocal<>("Current transaction isolation level");
     // 事务是否开启   actual：真实的
     private static final ThreadLocal<Boolean> actualTransactionActive = new NamedThreadLocal<>("Actual transaction active");
     }
     */

}